package com.chukun.flink.stream.connector;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;

import java.util.Properties;

/**
 * @author chukun
 * @version 1.0.0
 * @description kafka数据源设置
 * @createTime 2022年05月28日 18:28:00
 */
public class KafkaConnectorOperator {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        Properties conProperties = new Properties();
        conProperties.setProperty("bootstrap.servers", "linux01:9092");
        conProperties.setProperty("group.id", "kafka_flink_group");
        FlinkKafkaConsumer011<String> kafkaConsumer = new FlinkKafkaConsumer011<>(
                "kafka_flink", new SimpleStringSchema(), conProperties);

        DataStream<String> streamSource = env.addSource(kafkaConsumer);

        DataStream<String> mapStream = streamSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                int number = Integer.parseInt(value) + 1;
                Thread.sleep(100);
                System.out.println("number：" + number);
                return String.valueOf(number);
            }
        });


        Properties proProperties = new Properties();
        proProperties.put("bootstrap.servers", "linux01:9092");
        FlinkKafkaProducer011<String> kafkaProducer = new FlinkKafkaProducer011<>(
                "kafka_flink", new SimpleStringSchema(), proProperties);

        mapStream.addSink(kafkaProducer);

        DataStream<String> testSource = env.fromElements("1");

        testSource.addSink(kafkaProducer);

        env.execute("KafkaDataSource");
    }
}
